/**
 * 
 */
package com.sft.thriftMutiDemo.server;

import java.io.IOException;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;
import org.apache.thrift.TMultiplexedProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.server.ServerContext;
import org.apache.thrift.server.TServerEventHandler;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.server.TThreadPoolServer.Args;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;

import com.alibaba.fastjson.JSONObject;
import com.sft.service.Hello1Impl;
import com.sft.service.HelloImpl;
import com.sft.service.HelloWorldService;
import com.sft.service.HelloWorldService1;

/**
 * @author sft
 *
 *         2016年5月17日
 */
public class MutiServer {
	private static final Logger log = Logger.getLogger(MutiServer.class);
	private boolean isStart = Boolean.FALSE;
	private int port=1009;

	public static void main(String[] args) {
		MutiServer server = new MutiServer();
		server.startServer();
		// 一直等待，apache thrift启动完成
		synchronized (server) {
			try {
				while (!server.isStart) {
					server.wait();
				}
			} catch (InterruptedException e) {
				log.error(e);
				System.exit(-1);
			}
		}
		try {
			server.connectZk();
		} catch (IOException | KeeperException | InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	void connectZk() throws IOException, KeeperException, InterruptedException {
		log.info("===========开始连接zookeeper============");
		MyWatch watcher = new MyWatch();
		ZooKeeper zk = new ZooKeeper("192.168.2.92:2008", 120000, watcher);
		// 创建服务的父节点
		String pathName = "/thriftMuti";
		Stat path = null;
		try {
			path = zk.exists(pathName, watcher);
			if (path == null) {
				zk.create(pathName, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}

		} catch (KeeperException | InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			log.info("=========连接zookeeper异常，程序退出===========");
			System.exit(-1);
		}
		// 添加服务
		String serviceName = "/testThriftMuti";// 调用的服务名称
		JSONObject nodeJson = new JSONObject();
		nodeJson.put("ip", "192.168.2.92");
		nodeJson.put("port", port);
		zk.create(pathName + serviceName, nodeJson.toString().getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
	}

	void startServer() {
		new Thread() {
			@Override
			public void run() {

				// 基于阻塞式同步IO模型的Thrift服务，正式生产环境不建议用这个
				TServerSocket serverTransport = null;
				try {
					serverTransport = new TServerSocket(port);
				} catch (TTransportException e) {
					e.printStackTrace();
					System.exit(-1);
				}
				// 服务执行控制器（只要是调度服务的具体实现该如何运行）
				TMultiplexedProcessor processor = new TMultiplexedProcessor();
				processor.registerProcessor("hello", new HelloWorldService.Processor<com.sft.service.HelloWorldService.Iface>(new HelloImpl()));
				processor.registerProcessor("hello1", new HelloWorldService1.Processor<com.sft.service.HelloWorldService1.Iface>(new Hello1Impl()));
				// 为这个服务器设置对应的IO网络模型、设置使用的消息格式封装、设置线程池参数
				Args tArgs = new Args(serverTransport);
				tArgs.processor(processor);
				tArgs.protocolFactory(new TBinaryProtocol.Factory());
				tArgs.executorService(Executors.newFixedThreadPool(100));

				// 启动这个thrift服务
				TThreadPoolServer server = new TThreadPoolServer(tArgs);
				server.setServerEventHandler(new StartServerEventHandler());
				server.serve();
				
			}
		}.start();
	}

	 class StartServerEventHandler implements TServerEventHandler {
		@Override
		public void preServe() {
			/*
			 * 需要实现这个方法，以便在服务启动成功后， 通知mainProcessor： “Apache Thrift已经成功启动了”
			 */
			MutiServer.this.isStart = true;
			synchronized (MutiServer.this) {
				MutiServer.this.notify();
			}
		}

		/*
		 * (non-Javadoc)
		 * 
		 * @see
		 * org.apache.thrift.server.TServerEventHandler#createContext(org.apache
		 * .thrift.protocol.TProtocol, org.apache.thrift.protocol.TProtocol)
		 */
		@Override
		public ServerContext createContext(TProtocol input, TProtocol output) {
			/*
			 * 无需实现
			 */
			return null;
		}

		@Override
		public void deleteContext(ServerContext serverContext, TProtocol input, TProtocol output) {
			/*
			 * 无需实现
			 */
		}

		@Override
		public void processContext(ServerContext serverContext, TTransport inputTransport, TTransport outputTransport) {
			/*
			 * 无需实现
			 */
		}
	}

	private class MyWatch implements Watcher {

		/*
		 * (non-Javadoc)
		 * 
		 * @see org.apache.zookeeper.Watcher#process(org.apache.zookeeper.
		 * WatchedEvent)
		 */
		@Override
		public void process(WatchedEvent event) {
			// TODO Auto-generated method stub

		}

	}

}
